package com.yhou.cananl.client;

import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
import com.alibaba.otter.canal.protocol.CanalEntry;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author houyong
 * @ProjectName search
 * @Description:
 * @date 2018/09/05 下午5:47
 */
@Slf4j
public abstract class AbstractCanalClient {

    private static Logger log = LoggerFactory.getLogger(AbstractCanalClient.class);
//    @Value("${canal.zookeeper}")
//    public String zk;

    @Value("${canal.host}")
    public String canalhost;

    @Value("${canal.port}")
    public int canalport;

    @Value("${canal.username}")
    public String username;

    @Value("${canal.password}")
    public String password;

    private ExecutorService canalDeleteExecutor = Executors.newFixedThreadPool(10, new NamedThreadFactory("canal-delete"));

    /**
     * 单线程版本写入
     */
    public void handleCanalEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }
            CanalEntry.EventType eventType = rowChage.getEventType();
            log.info("================> binlog {}, {} , name {}, {} , eventType : {}",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType);

            if (eventType == CanalEntry.EventType.DELETE) {
                //删除事件
                deleteDoc(entry, rowChage);
            } else if (eventType == CanalEntry.EventType.INSERT) {
                createDoc(entry,rowChage);
            } else if (eventType == CanalEntry.EventType.UPDATE) {
                updateDoc(entry, rowChage);
            }
        }
    }


    /**
     * 多线程写入,写入异常，失败需要往上层抛异常。以便回滚binlog位点;更新单线程
     */

    public void handleCanalEntry(List<CanalEntry.Entry> entrys, ExecutorService executorService) {
        //只对insert事件使用
        List<Future<Boolean>> futures = new ArrayList<>();
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }
            CanalEntry.EventType eventType = rowChage.getEventType();
            log.info("================> binlog {}, {} , name {}, {} , eventType : {}",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType);
            if (eventType == CanalEntry.EventType.DELETE) {
                //删除事件
                deleteDoc(entry, rowChage);
            } else if (eventType == CanalEntry.EventType.INSERT) {
                //通一个batch有多条插入的情况下,多线程执行
                CreateDocTask createDocTask = new CreateDocTask(entry, rowChage);
                Future<Boolean> future = executorService.submit(createDocTask);
                futures.add(future);
            } else if (eventType == CanalEntry.EventType.UPDATE) {
                //update 多线程并发情况下,es基于乐观锁的并发控制机制,可能报version conflict
                //因此update 情况单线程串行执行
                // 只要有一条更新es异常,就往上层抛异常。以便回滚canal的位点
                updateDoc(entry, rowChage);
            }
        }

        for (Future future : futures) {
            try {
                Boolean result = (Boolean) future.get();
                if (!result) {
                    throw new RuntimeException("ERROR ## create index exception!!!");
                }
            } catch (InterruptedException e) {
                log.error("创建订单索引异步任务异常！", e);
                throw new RuntimeException("ERROR ## create index exception!!!");
            } catch (ExecutionException e) {
                log.error("创建订单索引异步任务异常！", e);
                throw new RuntimeException("ERROR ## create index exception!!!");
            }
        }
    }


    /**
     * 写入es 任务
     */
    class CreateDocTask implements Callable<Boolean> {

        private CanalEntry.Entry entry;
        private CanalEntry.RowChange rowChage;

        public CreateDocTask(CanalEntry.Entry entry, CanalEntry.RowChange rowChage) {
            this.entry = entry;
            this.rowChage = rowChage;
        }

        @Override
        public Boolean call() throws Exception {
            return createDoc(entry, rowChage);
        }
    }

    //创建文档
    private Boolean createDoc(CanalEntry.Entry entry, CanalEntry.RowChange rowChange) {
        boolean result = false;
        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            //如果创建索引失败,需要抛一个业务层异常到上层,这样canal会rollback 偏移量
            result = insertColumn(rowData.getAfterColumnsList(), entry);
            if (!result) { throw new RuntimeException("ERROR ## create index exception!!!");
            }
        }
        return result;
    }


    //更新文档
    private void updateDoc(CanalEntry.Entry entry, CanalEntry.RowChange rowChage) {
        for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
            boolean result = updateColumn(rowData.getAfterColumnsList(), entry);
            if (!result) {
                throw new RuntimeException("ERROR ## update es index exception , data:" + entry.toString());
            }
        }
    }

    //删除文档
    private void deleteDoc(CanalEntry.Entry entry, CanalEntry.RowChange rowChage) {
        CanalEntry.RowChange finalRowChage = rowChage;
        canalDeleteExecutor.execute(() -> {
            finalRowChage.getRowDatasList().forEach(rowData -> {
                try {
                    deleteColumn(rowData.getBeforeColumnsList(), entry);
                } catch (Exception e) {
                    log.error("ERROR ## delete index exception!!!data={}", entry, e);
                }
            });
        });
    }


    abstract boolean insertColumn(List<CanalEntry.Column> columns, CanalEntry.Entry entry);

    abstract boolean updateColumn(List<CanalEntry.Column> columns, CanalEntry.Entry entry);

    abstract boolean deleteColumn(List<CanalEntry.Column> columns, CanalEntry.Entry entry);


}
